package com.sh.data.engine.domain.datadev.flink.service.impl;

import cn.hutool.core.convert.Convert;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.google.common.collect.Lists;
import com.sh.data.engine.common.enumDefinition.FlinkNodeIconEnum;
import com.sh.data.engine.common.enumDefinition.FlinkNodeTopEnum;
import com.sh.data.engine.common.enumDefinition.FlinkNodeTypeEnum;
import com.sh.data.engine.common.enumDefinition.FlinkTopNodeIdEnum;
import com.sh.data.engine.common.exception.BusinessException;
import com.sh.data.engine.common.util.ServletUtils;
import com.sh.data.engine.domain.base.model.UserContext;
import com.sh.data.engine.domain.datadev.flink.model.domain.*;
import com.sh.data.engine.domain.datadev.flink.model.param.SaveTaskNodeParam;
import com.sh.data.engine.domain.datadev.flink.service.*;
import com.sh.data.engine.domain.util.ConvertUtil;
import com.sh.data.engine.repository.dao.datadev.FlinkMenuNodeMapper;
import com.sh.data.engine.repository.dao.datadev.entity.FlinkMenuNodeEntity;
import com.sh.data.engine.repository.dao.datadev.entity.FlinkNodeConfigEntity;
import com.sh.data.engine.repository.dao.datadev.entity.FlinkNodeConfigEntity.JarTaskConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.*;
import java.util.stream.Collectors;

/**
 * @author: zigui.zdf
 * @description:
 * @date: 2021/6/3 10:49
 */
@Service
@Slf4j
public class FlinkNodeServiceImpl extends ServiceImpl<FlinkMenuNodeMapper, FlinkMenuNodeEntity>
    implements FlinkNodeService {

    //  @Autowired private FlinkMenuNodeMapper flinkMenuNodeMapper;
    //  @Autowired private FlinkNodeConfigMapper flinkNodeConfigMapper;
    @Autowired
    private FlinkNodeConfigService flinkNodeConfigService;
    @Autowired
    private FlinkSqlTaskService flinkSqlTaskService;
    @Autowired
    private FlinkTaskService flinkTaskService;
    @Autowired
    private FlinkFunctionService flinkFunctionService;
    @Autowired
    private FlinkResourceService flinkResourceService;
    @Autowired
    private FlinkService flinkService;

    @Override
    public Boolean saveNode(FlinkNodeBaseDomain request, Long projectId, String userId) {
        Long pid = request.getPid();
        if (Objects.nonNull(pid)) {
            if (pid.equals(Long.valueOf(FlinkNodeTopEnum.FLINK_DEV_NODE.getId()))) {
                pid = Long.valueOf(FlinkNodeTopEnum.FLINK_DEV_NODE.getId());
            } else if (pid.equals(Long.valueOf(FlinkNodeTopEnum.FLINK_DEV_RESOURCE.getId()))) {
                pid = Long.valueOf(FlinkNodeTopEnum.FLINK_DEV_RESOURCE.getId());
            } else if (pid.equals(Long.valueOf(FlinkNodeTopEnum.FLINK_DEV_FUNCTION.getId()))) {
                pid = Long.valueOf(FlinkNodeTopEnum.FLINK_DEV_FUNCTION.getId());
            }
        }
        final Boolean existSameNodeName =
            this.hasSameMenuNode(
                request.getId(), pid, projectId, request.getName(), FlinkNodeTypeEnum.DIR.getCode());
        if (existSameNodeName) {
            throw new BusinessException("已存在同名目录或者节点");
        }
        FlinkMenuNodeEntity entity = new FlinkMenuNodeEntity();
        entity.setName(StringUtils.trim(request.getName()));
        entity.setPid(pid);
        entity.setUpdateTime(new Date());
        entity.setUpdaterId(userId);
        entity.setNodeType(request.getNodeType());
        entity.setRowState(1);
        Long nodeId = request.getId();
        entity.setRemark(request.getRemark());
        if (Objects.isNull(nodeId)) {
            entity.setCreateTime(new Date());
            entity.setNodeType(request.getNodeType());
            entity.setProjectId(projectId);
            entity.setCreatorId(userId);
            entity.setSort(this.getNextSort(projectId, pid));
            this.save(entity);
            nodeId = entity.getId();
        } else {
            entity.setId(request.getId());
            this.updateById(entity);
        }

        if (request.getNodeType().equals(FlinkNodeTypeEnum.TASK_JAR.getCode())) {
            FlinkNodeConfigDomain nodeConfigByNodeId = flinkNodeConfigService.getNodeConfigById(nodeId);
            boolean flag = true;
            if (nodeConfigByNodeId == null) {
                nodeConfigByNodeId = new FlinkNodeConfigDomain();
                nodeConfigByNodeId.setEnvConfig(flinkService.getDefaultEnv());
                nodeConfigByNodeId.setActive(true);
                flag = false;
            }
            JarTaskConfig jarTaskConfig = new JarTaskConfig();
            jarTaskConfig.setResourceNodeId(request.getResourceNodeId());
            jarTaskConfig.setMainClass(request.getMainClass());
            nodeConfigByNodeId.setNodeId(nodeId);
            nodeConfigByNodeId.setCreateTime(new Date());
            nodeConfigByNodeId.setUpdateTime(new Date());
            nodeConfigByNodeId.setCreatorId(userId);
            nodeConfigByNodeId.setUpdaterId(userId);
            FlinkNodeConfigEntity flinkNodeConfigEntity =
                ConvertUtil.copyProperties(nodeConfigByNodeId, FlinkNodeConfigEntity.class);
            flinkNodeConfigEntity.setJarTaskConfig(jarTaskConfig);
            if (flag) {
                flinkNodeConfigService.updateOne(flinkNodeConfigEntity);
            } else {
                flinkNodeConfigService.insertOne(flinkNodeConfigEntity);
            }
        }
        return true;
    }

    @Override
    public Boolean hasSameMenuNode(Long id, Long pid, Long projectId, String name, int nodeType) {

        LambdaQueryWrapper<FlinkMenuNodeEntity> lambdaQueryWrapper =
            new LambdaQueryWrapper<FlinkMenuNodeEntity>()
                .eq(FlinkMenuNodeEntity::getPid, pid)
                .eq(FlinkMenuNodeEntity::getName, name)
                .eq(FlinkMenuNodeEntity::getProjectId, projectId)
                .eq(FlinkMenuNodeEntity::getNodeType, nodeType);
        if (Objects.nonNull(id)) {
            lambdaQueryWrapper.ne(FlinkMenuNodeEntity::getId, id);
        }
        long count = count(lambdaQueryWrapper);

        if (count > 0) {
            return true;
        }
        return false;
    }

    @Override
    public List<FlinkNodeDomain> getAllNodes(Integer type, Long projectId) {
        String pid;
        if (type == 0) {
            pid = "1";
        } else if (type == 1) {
            pid = "2";
        } else {
            pid = "3";
        }
        List<FlinkNodeDomain> nodes = Lists.newArrayList();
        final List<FlinkMenuNodeEntity> entities = Lists.newLinkedList();
        this.getNodesByPid(Long.valueOf(pid), projectId, false, entities);
        for (FlinkMenuNodeEntity node : entities) {
            final FlinkNodeDomain flinkNodeDomain = Convert.convert(FlinkNodeDomain.class, node);
            flinkNodeDomain.setId(node.getId());
            flinkNodeDomain.setIcon(FlinkNodeIconEnum.getIconByNodeType(node.getNodeType()));

            nodes.add(flinkNodeDomain);
        }
        return nodes;
    }

    @Override
    public FlinkNodeDomain getFinkNodeById(Long id) {
        FlinkMenuNodeEntity entity = this.getById(id);
        FlinkNodeDomain nodeDomain = Convert.convert(FlinkNodeDomain.class, entity);
        return nodeDomain;
    }

    @Override
    @Transactional(rollbackFor = Exception.class, transactionManager = "engineTransactionManager")
    public void deleteNode(Long id, Long projectId) {
        FlinkMenuNodeEntity entity = this.getById(id);

        if (Objects.isNull(entity)) {
            throw new BusinessException("数据不存在");
        }

        final Integer nodeType = entity.getNodeType();

        if (Objects.equals(nodeType, FlinkNodeTypeEnum.DIR.getCode())) {
            deleteDir(id, projectId);
            return;
        }
        this.removeById(id);
        flinkNodeConfigService.deleteByNodeId(id);
    }

    @Override
    public List<FlinkDirTreeDomain> getDirNodes(Integer type, Long projectId) {
        String pid;
        if (type == 0) {
            pid = "1";
        } else if (type == 1) {
            pid = "2";
        } else {
            pid = "3";
        }

        List<FlinkDirTreeDomain> result = Lists.newArrayList();
        List<FlinkMenuNodeEntity> nodes = Lists.newLinkedList();
        this.getNodesByPid(Long.valueOf(pid), projectId, true, nodes);
        FlinkDirTreeDomain domain = new FlinkDirTreeDomain();
        FlinkTopNodeIdEnum topNodeIdEnum = FlinkTopNodeIdEnum.of(pid);
        domain.setId(Long.valueOf(topNodeIdEnum.getId()));
        domain.setName(topNodeIdEnum.getName());
        getChildDirs(Long.valueOf(pid), nodes, domain);
        result.add(domain);
        //    for (FlinkMenuNodeEntity node : entities) {
        //      final FlinkNodeDomain flinkNodeDomain = Convert.convert(FlinkNodeDomain.class, node);
        //      flinkNodeDomain.setId(String.valueOf(node.getId()));
        //      flinkNodeDomain.setIcon(FlinkNodeIconEnum.getIconByNodeType(node.getNodeType()));
        //      nodes.add(flinkNodeDomain);
        //    }
        return result;
    }

    private void getChildDirs(Long pid, List<FlinkMenuNodeEntity> nodes, FlinkDirTreeDomain domain) {
        List<FlinkDirTreeDomain> childs = new ArrayList<>();

        nodes.stream()
            .filter(node -> Objects.equals(node.getPid(), pid))
            .forEach(
                node -> {
                    FlinkDirTreeDomain child = new FlinkDirTreeDomain();
                    child.setId(node.getId());
                    child.setName(node.getName());
                    child.setPid(node.getPid());
                    child.setNodeType(node.getNodeType());
                    child.setSort(node.getSort());
                    childs.add(child);

                    getChildDirs(node.getId(), nodes, child);
                });

        domain.setChilds(childs);
    }

    @Override
    public Long saveFunctionNode(FlinkNodeFunctionDomain request) {
        Long projectId = ServletUtils.getProjectId();
        String userId = UserContext.getUserId();

        final Boolean existSameNodeName =
            this.existSameFunctionNode(request.getNodeId(), request.getNodeName(), projectId);
        if (existSameNodeName) {
            throw new BusinessException("已存在同名函数");
        }

        SaveTaskNodeParam param = new SaveTaskNodeParam();
        param.setId(request.getNodeId());
        param.setPid(request.getPid());
        param.setName(request.getNodeName());
        param.setUserId(userId);
        param.setProjectId(projectId);
        param.setNodeType(FlinkNodeTypeEnum.FUNCTION.getCode());
        final Long nodeId = this.saveNode(param);
        flinkNodeConfigService.saveFunctionConfig(
            nodeId,
            request.getFunctionType(),
            request.getClassName(),
            request.getResourceNodeId(),
            request.getPurpose(),
            request.getFormat(),
            request.getParamDesc(),
            userId,
            projectId);
        return nodeId;
    }

    @Override
    public boolean existSameFunctionNode(Long id, String name, Long projectId) {

        LambdaQueryWrapper<FlinkMenuNodeEntity> eq =
            new LambdaQueryWrapper<FlinkMenuNodeEntity>()
                .eq(FlinkMenuNodeEntity::getName, name)
                .eq(FlinkMenuNodeEntity::getNodeType, FlinkNodeTypeEnum.FUNCTION.getCode())
                .eq(FlinkMenuNodeEntity::getProjectId, projectId);
        if (Objects.nonNull(id)) {
            eq.ne(FlinkMenuNodeEntity::getId, id);
        }
        long count = this.count(eq);
        if (count > 0) {
            return true;
        }
        return false;
    }

    @Override
    public boolean existSameNode(Long id, String name, Long pid, Long projectId) {

        LambdaQueryWrapper<FlinkMenuNodeEntity> eq =
            new LambdaQueryWrapper<FlinkMenuNodeEntity>()
                .eq(FlinkMenuNodeEntity::getName, name)
                .eq(FlinkMenuNodeEntity::getPid, pid)
                .eq(FlinkMenuNodeEntity::getProjectId, projectId);
        if (Objects.nonNull(id)) {
            eq.ne(FlinkMenuNodeEntity::getId, id);
        }
        long count = this.count(eq);
        if (count > 0) {
            return true;
        }
        return false;
    }

    @Override
    public List<FlinkResourceListDomain> getResourceByProjectId(Long projectId) {
        List<FlinkNodeDomain> allNodes = getAllNodes(1, projectId);

        List<Long> nodeIds =
            allNodes.stream()
                .filter(i -> !i.getNodeType().equals(FlinkNodeTypeEnum.DIR))
                .map(i -> i.getId())
                .collect(Collectors.toList());

        List<FlinkNodeConfigEntity> byNodeIds = flinkNodeConfigService.getByNodeIds(nodeIds);

        Map<Long, String> resourceMap =
            byNodeIds.stream()
                .filter(i -> i.getIsActive().equals(1))
                .collect(
                    Collectors.toMap(
                        FlinkNodeConfigEntity::getId,
                        i ->
                            Objects.isNull(i.getResourceConfig())
                                || StringUtils.isBlank(i.getResourceConfig().getResourceName())
                                ? ""
                                : i.getResourceConfig().getResourceName()));
        Set<Map.Entry<Long, String>> entries = resourceMap.entrySet();
        List<FlinkResourceListDomain> result = Lists.newArrayList();
        entries.forEach(
            i -> {
                FlinkResourceListDomain flinkResourceListDomain = new FlinkResourceListDomain();
                flinkResourceListDomain.setConfigId(i.getKey());
                flinkResourceListDomain.setName(i.getValue());
                result.add(flinkResourceListDomain);
            });
        return result;
    }

    @Override
    public Boolean lockNode(Long id, Integer isLocked) {
        LambdaUpdateWrapper<FlinkMenuNodeEntity> update =
            new UpdateWrapper<FlinkMenuNodeEntity>().lambda();
        update.eq(FlinkMenuNodeEntity::getId, id).set(FlinkMenuNodeEntity::getIsLocked, isLocked);
        this.update(update);
        return Boolean.TRUE;
    }

    @Override
    public Boolean publish(Long id, Integer status, String userId) {
        final FlinkMenuNodeEntity entity = this.getById(id);

        final Integer nodeType = entity.getNodeType();

        final FlinkNodeTypeEnum nodeTypeEnum = FlinkNodeTypeEnum.getByNodeType(nodeType);

        switch (nodeTypeEnum) {
            case TASK_SQL:
                flinkSqlTaskService.publish(id, status, userId, entity.getRemark());
                break;
            case TASK_JAR:
                flinkTaskService.publish(id, status, userId, entity.getRemark());
                break;
            case FUNCTION:
                flinkFunctionService.publish(id, status, userId);
                break;
            case RESOURCE:
                flinkResourceService.publish(id, status, userId);
                break;
        }

        return Boolean.TRUE;
    }

    @Override
    public List<FlinkNodeDomain> getListByPage(Long projectId, Integer pageSize, Integer pageNum) {

        LambdaQueryWrapper<FlinkMenuNodeEntity> eq =
            new LambdaQueryWrapper<FlinkMenuNodeEntity>()
                .eq(FlinkMenuNodeEntity::getProjectId, projectId)
                .eq(FlinkMenuNodeEntity::getRowState, 1);
        IPage<FlinkMenuNodeEntity> page = page(new Page<>(pageNum, pageSize), eq);

        return ConvertUtil.copyProperties(page.getRecords(), FlinkNodeDomain.class);
    }

    @Override
    public List<FlinkNodeConfigResourceDomain> getAllActiveResource(Long projectId) {

        List<FlinkNodeConfigEntity> resourceEntity =
            flinkNodeConfigService.getActiveResourceByProjectId(projectId);
        List<FlinkNodeConfigResourceDomain> domains =
            resourceEntity.stream()
                .map(
                    r -> {
                        FlinkNodeConfigResourceDomain convert =
                            Convert.convert(FlinkNodeConfigResourceDomain.class, r);
                        convert.setFileName(r.getResourceConfig().getFileName());
                        convert.setFilePath(r.getResourceConfig().getPath());
                        convert.setRemark(r.getResourceConfig().getRemark());
                        return convert;
                    })
                .collect(Collectors.toList());
        return domains;
    }

    @Override
    public List<FlinkMenuSimpleDomain> getFlinkMenuSimpleList(Long projectId) {
        LambdaQueryWrapper<FlinkMenuNodeEntity> eq =
            new LambdaQueryWrapper<FlinkMenuNodeEntity>()
                .eq(FlinkMenuNodeEntity::getProjectId, projectId)
                .eq(FlinkMenuNodeEntity::getIsPublished, 1)
                .eq(FlinkMenuNodeEntity::getNodeType, FlinkNodeTypeEnum.TASK_SQL.getCode());
        List<FlinkMenuNodeEntity> multi = this.list(eq);

        List<FlinkMenuSimpleDomain> flinkMenuSimpleDomains =
            ConvertUtil.copyProperties(multi, FlinkMenuSimpleDomain.class);
        return flinkMenuSimpleDomains;
    }

    @Override
    public boolean changeNodeDir(Long id, Long pid) {

        Long projectId = ServletUtils.getProjectId();
        String userId = UserContext.getUserId();

        final FlinkNodeDomain target = this.getFinkNodeById(id);
        final FlinkNodeDomain parent = this.getFinkNodeById(pid);

        if (!parent.getNodeType().equals(FlinkNodeTypeEnum.DIR.getCode())) {
            throw new BusinessException("不支持的操作");
        }

        if (target.isSystem()) {
            throw new BusinessException("不支持的操作");
        }

        if (Objects.equals(target.getPid(), pid)) {
            throw new BusinessException("已在当前目录中");
        }
        final Integer nextSort = this.getNextSort(projectId, pid);
        FlinkMenuNodeEntity entity = this.getById(id);
        entity.setPid(pid);
        entity.setSort(nextSort);
        entity.setUpdaterId(userId);
        entity.setUpdateTime(new Date());
        this.updateById(entity);
        return true;
    }

    @Override
    public Integer getNextSort(Long projectId, Long pid) {

        //        .addCriteria(Criteria.where("pid").is(pid))
        //        .addCriteria(Criteria.where("projectId").is(projectId))
        //        .limit(1);
        //    final Sort sort = Sort.by(Sort.Direction.DESC, "sort");
        //    final List<FlinkMenuNodeEntity> entityList = this.findMulti(query, sort);
        //    if (CollectionUtils.isNotEmpty(entityList)) {
        //      return entityList.get(0).getSort();
        //    }
        LambdaQueryWrapper<FlinkMenuNodeEntity> lambdaQueryWrapper =
            new LambdaQueryWrapper<FlinkMenuNodeEntity>()
                .eq(FlinkMenuNodeEntity::getPid, pid)
                .eq(FlinkMenuNodeEntity::getProjectId, projectId)
                .last("limit 1")
                .orderByDesc(FlinkMenuNodeEntity::getSort);
        FlinkMenuNodeEntity one = this.getOne(lambdaQueryWrapper, false);
        Integer maxSort;
        if (one == null) {
            maxSort = null;
        } else {
            maxSort = one.getSort();
        }
        maxSort = Objects.isNull(maxSort) ? 0 : maxSort;
        // Optional.ofNullable(maxSort).orElse(0)
        return maxSort + 1;
    }

    @Override
    public boolean changeNodeSort(Long id, Long siblingsId, Integer dropPosition) {
        Long projectId = ServletUtils.getProjectId();

        final FlinkNodeDomain target = this.getFinkNodeById(id);
        final FlinkNodeDomain siblings = this.getFinkNodeById(siblingsId);

        if (target.isSystem()) {
            throw new BusinessException("不支持的操作");
        }

        Long targetPid = target.getPid();
        Integer targetSort;
        Integer siblingsSort = siblings.getSort();

        if (!Objects.equals(target.getPid(), siblings.getPid())) {
            final Boolean existSameNodeName =
                this.existSameNode(null, target.getName(), siblings.getPid(), projectId);
            if (existSameNodeName) {
                throw new BusinessException("已存在同名节点");
            }
            targetPid = siblings.getPid();
        }

    /* // target在siblings的上面
    if (dropPosition == -1) {
      targetSort = siblingsSort;
      flinkMenuNodeService.batchIncreaseNodeSort(targetPid, siblingsSort - 1);
    } else {
      targetSort = siblingsSort + 1;
      flinkMenuNodeService.batchIncreaseNodeSort(targetPid, siblingsSort);
    }

    flinkMenuNodeService.setNodePidAndSort(target.getId(), targetPid, targetSort, userId);*/
        return true;
    }

    @Override
    public Long saveNode(SaveTaskNodeParam param) {
        String name = StringUtils.trim(param.getName());

        FlinkMenuNodeEntity entity = new FlinkMenuNodeEntity();
        entity.setName(name);
        entity.setPid(param.getPid());
        entity.setUpdateTime(new Date());
        entity.setUpdaterId(param.getUserId());
        entity.setRemark(param.getRemark());
        entity.setDevType(param.getDevType());
        if (param.getId() == null) {
            entity.setCreateTime(new Date());
            entity.setIsSystem(0);
            entity.setProjectId(param.getProjectId());
            entity.setCreatorId(param.getUserId());
            entity.setIsLocked(0);
            entity.setNodeType(param.getNodeType());
            entity.setSort(this.getNextSort(param.getProjectId(), param.getPid()));
            this.save(entity);
            return entity.getId();
        }
        entity.setId(param.getId());
        this.updateById(entity);
        return param.getId();
    }

    private void getNodesByPid(
        Long pid, Long projectId, boolean onlyDirNode, List<FlinkMenuNodeEntity> list) {
        final List<FlinkMenuNodeEntity> entities = this.getNodesByPidDB(projectId, pid, onlyDirNode);
        list.addAll(entities);
        for (FlinkMenuNodeEntity entity : entities) {
            if (entity.getNodeType().equals(FlinkNodeTypeEnum.DIR.getCode())) {
                getNodesByPid(entity.getId(), projectId, onlyDirNode, list);
            }
        }
    }

    private List<FlinkMenuNodeEntity> getNodesByPidDB(Long projectId, Long pid, boolean onlyDirNode) {
        LambdaQueryWrapper<FlinkMenuNodeEntity> eq =
            new LambdaQueryWrapper<FlinkMenuNodeEntity>()
                .eq(FlinkMenuNodeEntity::getPid, pid)
                .eq(FlinkMenuNodeEntity::getProjectId, projectId);
        if (onlyDirNode) {
            eq.eq(FlinkMenuNodeEntity::getNodeType, FlinkNodeTypeEnum.DIR.getCode());
        }
        eq.orderByDesc(FlinkMenuNodeEntity::getSort);
        List<FlinkMenuNodeEntity> list = list(eq);
        return list;
    }

    private void deleteDir(Long id, Long projectId) {
        final List<FlinkMenuNodeDomain> domains = this.recursiveGetAllChildNodes(projectId, id, false);
        if (CollectionUtils.isNotEmpty(domains)) {
            throw new BusinessException("非空目录，无法tb_md_datadistribution_table删除");
        }

        List<FlinkMenuNodeDomain> dirNodeDomains = this.recursiveGetAllChildNodes(projectId, id, true);
        List<Long> dirNodeIds =
            dirNodeDomains.stream().map(FlinkMenuNodeDomain::getId).collect(Collectors.toList());
        dirNodeIds.add(id);
        this.removeBatchByIds(dirNodeIds);
    }

    //  @Override
    public List<FlinkMenuNodeDomain> recursiveGetAllChildNodes(
        Long projectId, Long pid, boolean includeDir) {
        List<FlinkMenuNodeDomain> domains = Lists.newArrayList();
        getByPid(projectId, pid, includeDir, domains);
        return domains;
    }

    private void getByPid(
        Long projectId, Long pid, boolean includeDir, List<FlinkMenuNodeDomain> domains) {

        LambdaQueryWrapper<FlinkMenuNodeEntity> eq =
            new LambdaQueryWrapper<FlinkMenuNodeEntity>()
                .eq(FlinkMenuNodeEntity::getProjectId, projectId)
                .eq(FlinkMenuNodeEntity::getPid, pid);
        final List<FlinkMenuNodeEntity> entities = this.list(eq);
        for (FlinkMenuNodeEntity entity : entities) {
            if (Objects.equals(entity.getNodeType(), FlinkNodeTypeEnum.DIR.getCode())) {
                getByPid(projectId, entity.getId(), includeDir, domains);
                if (includeDir) {
                    domains.add(Convert.convert(FlinkMenuNodeDomain.class, entity));
                }
            } else {
                domains.add(Convert.convert(FlinkMenuNodeDomain.class, entity));
            }
        }
    }
}
